import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.netty.tcp.TcpClient;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

public class RSocketClient {

    private static final Logger logger = LoggerFactory.getLogger(RSocketClient.class);

    public static void main(String[] args) {
        TcpClientTransport transport = TcpClientTransport.create(TcpClient.create().port(18000).wiretap(false));
        RSocket socket =
                RSocketConnector.connectWith(transport).block();

        /*socket.requestChannel(
                Flux.interval(Duration.ofMillis(1000)).map(i -> DefaultPayload.create("Hello")))
                .map(Payload::getDataUtf8)
                .doOnNext(logger::debug)
                .take(10)
                .doFinally(signalType -> socket.dispose())
                .then()
                .block();*/

        AtomicInteger c = new AtomicInteger(0);
        Flux.interval(Duration.ofMinutes(1))
                .subscribe(l -> {
                    System.out.println("Tps:" + (c.get() / 60));
                    c.set(0);
                });

        Scheduler scheduler = Schedulers.newSingle((r) -> new Thread(r));

        while (true) {
            if (!socket.isDisposed()) {
                socket.requestResponse(DefaultPayload.create("Hello"))
                        .publishOn(scheduler)
                        .subscribeOn(scheduler)
                        .subscribe(payload -> {
                            System.out.println(Thread.currentThread().getName());
                            c.incrementAndGet();
                        });
            }
        }


    }
}
